/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.cluster.impl;

import org.sabayframework.activity.Activity;
import org.sabayframework.activity.ActivityMonitor;
import org.sabayframework.channel.Channel;
import org.sabayframework.cluster.Cluster;
import org.sabayframework.cluster.ClusterContext;
import org.sabayframework.cluster.DistributedActivity;
import org.sabayframework.cluster.NodeExecutionQueue;

/**
 * Implementation of NodeExecutionQueue which submits activities to an
 * execution queue on specified remote computer.
 * */
public class RemoteExecutionQueue implements NodeExecutionQueue {

	private final RemoteComputer remoteComputer;
	
	public RemoteExecutionQueue(RemoteComputer remoteComputer) {
		super();
		this.remoteComputer = remoteComputer;
	}

	@Override
	public <I, O> ActivityMonitor<I, O> submit(Activity<I, O> activity,
			Channel<I> input, Channel<O> output) {
		DistributedActivity<I, O> da= new DistributedActivity<I, O>(
				activity, 
				new SingleNodeStrategy(remoteComputer.getId()));
		Cluster cluster = ClusterContext.getClusteringService().getCluster();
		return cluster.getExecutionQueue().submit(da, input, output);
	}

	@Override
	public <I, O> ActivityMonitor<I, O> execute(Activity<I, O> activity,
			Channel<I> input, Channel<O> output) {
		return submit(activity, input, output);
	}

	@Override
	public int workerCount() {
		return remoteComputer.getChildrenCount();
	}

	@Override
	public int size() {
		return 0;
	}

	@Override
	public boolean isEmpty() {
		return true;
	}

	@Override
	public void channelChanged(Channel<?> input) {
	}

	@Override
	public void stop() {
	}

	@Override
	public <I, O> ActivityMonitor<I, O> submit(
			DistributedActivity<I, O> activity, Channel<I> input,
			Channel<O> output) {
		DistributedActivity<I, O> da= new DistributedActivity<I, O>(
				activity.getActivity(), 
				new SingleNodeStrategy(remoteComputer.getId()),
				activity.getImportPackage(),
				activity.getRequireBundle());
		
		Cluster cluster = ClusterContext.getClusteringService().getCluster();
		return cluster.getExecutionQueue().submit(da, input, output);
	}

}
